package com.iqiyi.pps.epg.web.timer;

import com.alibaba.fastjson.JSONObject;
import com.iqiyi.kiwi.utils.HttpClientConnection;
import com.iqiyi.pps.epg.core.model.channel.ChannelBase;
import com.iqiyi.pps.epg.core.model.channel.ChannelConfig;
import com.iqiyi.pps.epg.core.model.control.ProgramPlatformControl;
import com.iqiyi.pps.epg.core.model.program.ProgramData;
import com.iqiyi.pps.epg.core.model.stream.LiveStream;
import com.iqiyi.pps.epg.core.model.stream.StreamSpanLog;
import com.iqiyi.pps.epg.core.service.broadcast.BroadcastControlService;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.program.ProgramService;
import com.iqiyi.pps.epg.core.service.stream.LiveStreamService;
import com.iqiyi.pps.epg.core.utils.DateTimeUtil;
import com.iqiyi.pps.epg.core.utils.MailContentUtil;
import com.iqiyi.pps.epg.core.utils.SendMailUtils;
import com.iqiyi.pps.epg.core.utils.SendMsgUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class LiveStreamStatusCheckTrigger {
    private static Logger logger = LoggerFactory.getLogger(LiveStreamStatusCheckTrigger.class);

    @Autowired
    private LiveStreamService liveStreamService;
    @Autowired
    private ChannelService channelService;
    @Autowired
    private ProgramService programService;
    @Autowired
    private BroadcastControlService broadcastControlService;

    private ConcurrentLinkedQueue<LiveStream> queue;
    private ConcurrentHashMap<LiveStream, String> poors;

    private final static String TEMPLATE_SPAN_MSG = "%s<<%s>> %s 异常: %s       %s [%s]";

    public void task() {
        logger.info("[LiveStreamStatusCheckTrigger][task][start]");
        List<LiveStream> results = liveStreamService.getNeedCheckList();
        if (results.size() > 0) {
            poors = new ConcurrentHashMap();
            queue = new ConcurrentLinkedQueue<LiveStream>(results);
            ExecutorService exec = Executors.newFixedThreadPool(3);
            for (int i = 0; i < 3; i++) {
                exec.execute(new Executor());
            }
            exec.shutdown();
            try {
                while (!exec.awaitTermination(30, TimeUnit.SECONDS)) ;
            } catch (InterruptedException e) {
                logger.info("[LiveStreamStatusCheckTrigger][task][awaitTermination][exception={}]", e);
            }
            // 发送报警邮件
            if (poors.size() > 0) {
                StringBuilder sb = new StringBuilder("以下流启动状态异常<br /><br />");
                List<LiveStream> lsLivestreamList = new ArrayList<>();
                for (Map.Entry<LiveStream, String> item : poors.entrySet()) {
                    ChannelBase channelBase = channelService.getChannelBase(item.getKey().getFatherId());

                    if (channelBase.getFormal() == 0) {
                        continue;
                    }

                    sb.append("流格式: \t").append(item.getKey().getFormatType()).append("<br />")
                            .append("频道名称: \t").append(channelBase.getName()).append("<br />")
                            .append("频道ID：\t").append(item.getKey().getFatherId()).append("<br />")
                            .append("奇谱ID：\t").append(channelBase.getQipuId()).append("<br />")
                            .append("&nbsp;&nbsp;")
                            .append("<a href='")
                            .append(getStatUrl(item.getKey().getFormatType(), item.getKey().getUrl(), item.getKey().getDolby_ch()))
                            .append("' target='_blank'>")
                            .append(item.getKey().getUrl())
                            .append("</a>&nbsp;&nbsp;&nbsp;&nbsp;")
                            .append(item.getValue())
                            .append("<br />");

                    if (item.getKey().getChannelLiveType() == 2) {
                        item.getKey().setStreamSpan(item.getValue());
                        lsLivestreamList.add(item.getKey());
                    }

                }
                poors.clear(); // 清空本次数据
                List<ChannelConfig> mailList = channelService.getConfig(ChannelConfig.TEAM_TRIGGER_MAIL_LIST);
                String[] tos = new String[mailList.size()];
                int i = 0;
                for (ChannelConfig item : mailList) {
                    tos[i] = item.getVal();
                    ++i;
                }

                SendMailUtils.send("直播流状态检测任务报警", tos, sb.toString());

                List<String> msgList = new ArrayList<>();
                for (LiveStream temp : lsLivestreamList) {
                    String msg = checkLiveProgram(temp);
                    if (null != msg) {
                        msgList.add(msg);
                    }
                }

                if (msgList.size() > 0) {
                    String msgToList = getMsgToList();
                    for (String msg : msgList) {
                        SendMsgUtils.sendMsg(msgToList, msg, "", "warn");
                    }
                }
            }
        }
    }

    private String getMsgToList() {
        List<ChannelConfig> configs = channelService.getConfig(ChannelConfig.STREAM_SPAN_MSG_GROUP);
        StringBuffer sb = new StringBuffer();
        int size = configs.size();
        for (int i = 0; i < size; i++) {
            if (i > 0) {
                sb.append(",");
            }
            sb.append(configs.get(i).getVal());
        }

        return sb.toString();
    }

    private String checkLiveProgram(LiveStream livestream) {
        ProgramData currentLiveProgram = getCurrentLiveProgram(livestream.getFatherId());
        if (null != currentLiveProgram && hasBroadcaseOnline(currentLiveProgram.getId())) {
            return String.format(TEMPLATE_SPAN_MSG, DateTimeUtil.transOnlyTimeFormat(currentLiveProgram.getStartPlayTime().getTime()),
                    currentLiveProgram.getPname(), currentLiveProgram.getQipuId(), MailContentUtil.transToStreamType(livestream.getStreamType()),
                    livestream.getStreamSpan(), livestream.getFormatType());
        }
        return null;
    }

    private ProgramData getCurrentLiveProgram(long channelId) {
        List<ProgramData> programDataList = programService.getCurrentProgram(Long.valueOf(channelId).intValue());
        return programDataList != null && programDataList.size() > 0 ? programDataList.get(0) : null;
    }

    private boolean hasBroadcaseOnline(long programId) {
        int count = 0;
        List<ProgramPlatformControl> list_program = broadcastControlService.getProgramPlatformControls(programId);
        for (ProgramPlatformControl temp : list_program) {
            if (temp.getAvailableStatus() == 1) {
                count++;
            }
        }

        return count > 0;
    }

    private boolean saveStateInfo(JSONObject stateInfo, LiveStream liveStream) {
        try {
            StreamSpanLog streamSpanLog = new StreamSpanLog();
            streamSpanLog.setFatherId(liveStream.getFatherId());
            streamSpanLog.setStreamName(liveStream.getStreamName());
            streamSpanLog.setStreamType(liveStream.getFormatType());
            streamSpanLog.setChannelName(stateInfo.getString("channel_name"));
            streamSpanLog.setRestartTime(stateInfo.getString("restart_time"));
            streamSpanLog.setUpdateTime(stateInfo.getString("update_time"));
            streamSpanLog.setFirstSlice(stateInfo.getString("first_slice"));
            streamSpanLog.setMiniChunkId(stateInfo.getString("mini_chunk_id"));
            streamSpanLog.setCurrentChunkId(stateInfo.getString("current_chunk_id"));
            streamSpanLog.setSpan(stateInfo.getLong("span"));
            streamSpanLog.setSeq(stateInfo.getString("seq"));
            streamSpanLog.setAddTime(new Timestamp(System.currentTimeMillis()));

            liveStreamService.saveStreamSpanLog(streamSpanLog);
            return true;
        } catch (Exception e) {
            logger.info("[LiveStreamStatusCheckTrigger][saveStateInfo] exception=" + e);
        }
        return false;
    }

    private String getStatUrl(String formatType, String streamUrl, String dolbyCh) {
        String url = "http://cutteradmin.qiyi.domain:7777/";
        // MP4和TS检测一样
        if (formatType.equals("TS")) {
            if(StringUtils.isBlank(dolbyCh)) {
                url += "statm3u8_ts?ch=" + streamUrl + "&user=ppslive&src_type=PPC&version=2.0";
            }else{
                url += "stathttp2ts?ch=" + dolbyCh + "&user=ppslive&src_type=PPC&version=2.0";
            }
        } else {
            url += "statchannel?ch=" + streamUrl + "&user=ppslive&src_type=PPC&version=2.0";
        }
        return url;
    }

    public class Executor implements Runnable {
        @Override
        public void run() {
            LiveStream entity;
            while (true) {
                entity = null;
                synchronized (queue) {
                    if (!queue.isEmpty()) {
                        entity = queue.poll();
                    }
                }

                // 无任务退出，HCDN和FLV一起不需要检测，P2P无法检测，MP4和TS一起就不需要
//                if (entity == null || entity.getFormatType().equals("HCDN")
//                    || entity.getFormatType().equals("P2P") || entity.getFormatType().equals("MP4")) {
//                    break;
//                }
                // 无任务退出
                if (entity == null) {
                    break;
                }

                String url = getStatUrl(entity.getFormatType(), entity.getUrl(), entity.getDolby_ch());
                HttpClientConnection hcc = new HttpClientConnection(url, HttpClientConnection.GET_METHOD);
                for (int i = 0; i < 3; i++) {
                    hcc.connect();
                    if (hcc.getStatus() == 200) {
                        String result = hcc.getBody();
                        // 判断span > 20
                        JSONObject jsonResult = JSONObject.parseObject(result);

                        if (jsonResult.getIntValue("status") == 1) {
                            JSONObject jsonData = jsonResult.getJSONObject("data");
                            saveStateInfo(jsonData, entity);
                            int span = jsonData.getIntValue("span");
                            if (span > 60) {
                                poors.put(entity, "span = " + jsonData.getString("span"));
                            }
                            break;
                        } else {
                            poors.put(entity, result); // 接口异常
                        }
                    }
                }
            }
        }
    }
}
